package com.example.wan1.bus;

import org.reactivestreams.Subscription;

import java.util.concurrent.atomic.AtomicReference;

import io.reactivex.FlowableSubscriber;
import io.reactivex.disposables.Disposable;
import io.reactivex.exceptions.CompositeException;
import io.reactivex.exceptions.Exceptions;
import io.reactivex.functions.Action;
import io.reactivex.functions.Consumer;
import io.reactivex.internal.functions.Functions;
import io.reactivex.internal.subscriptions.SubscriptionHelper;
import io.reactivex.observers.LambdaConsumerIntrospection;
import io.reactivex.plugins.RxJavaPlugins;

/**
 * <pre>
 *     author: Blankj
 *     blog  : http://blankj.com
 *     time  : 2018/05/08
 *     desc  : 仿照LambdaObserver写的
 * </pre>
 * 1.AtomicReference:可以保证你在修改对象引用时的线程安全性
 * 2.serialVersionUID
 * 3.Exceptions.throwIfFatal(ex);
 * 4.LambdaConsumerIntrospection
 * 5.Subscription
 * <p>
 * get()是并发里的
 */

final class MyLambdaSubscriber<T> extends AtomicReference<Subscription>
        implements FlowableSubscriber<T>, Subscription, Disposable, LambdaConsumerIntrospection {

    private static final long serialVersionUID = -7251123623727029452L;
    final Consumer<? super T> onNext;
    final Consumer<? super Throwable> onError;
    final Action onComplete;
    final Consumer<? super Subscription> onSubscribe;

    /***
     * 构造方法 初始化数据
     * @param onNext 传递事件
     * @param onError 传递错误
     * @param onComplete 传递完成
     * @param onSubscribe 订阅
     */
    public MyLambdaSubscriber(Consumer<? super T> onNext, Consumer<? super Throwable> onError,
                              Action onComplete,
                              Consumer<? super Subscription> onSubscribe) {
        super();
        this.onNext = onNext;
        this.onError = onError;
        this.onComplete = onComplete;
        this.onSubscribe = onSubscribe;
    }

    //------------------------------FlowableSubscriber-------------------------------------
    //FlowableSubscriber继承自Subscriber，onSubscribe()方法是非空的

    /***
     * FlowableSubscriber
     * @param s
     */
    @Override
    public void onSubscribe(Subscription s) {
        if (SubscriptionHelper.setOnce(this, s)) {
            try {
                onSubscribe.accept(this);
            } catch (Throwable ex) {
                Exceptions.throwIfFatal(ex);
                s.cancel();
                onError(ex);
            }
        }
    }

    /***
     * 发送事件，遇到错误的处理
     * @param t
     */
    @Override
    public void onNext(T t) {
        //如果没有切断
        if (!isDisposed()) {
            try {
                onNext.accept(t);
            } catch (Throwable e) {
                Exceptions.throwIfFatal(e);
//                get().cancel();
                onError(e);
            }
        }
    }

    /**
     * 错误
     *
     * @param t
     */
    @Override
    public void onError(Throwable t) {
        if (get() != SubscriptionHelper.CANCELLED) {
//            lazySet(SubscriptionHelper.CANCELLED);
            try {
                onError.accept(t);
            } catch (Throwable e) {
                //
                Exceptions.throwIfFatal(e);
                RxJavaPlugins.onError(new CompositeException(t, e));
            }
        } else {
            RxJavaPlugins.onError(t);
        }
    }

    /**
     * 完成
     */
    @Override
    public void onComplete() {
        if (get() != SubscriptionHelper.CANCELLED) {
            lazySet(SubscriptionHelper.CANCELLED);
            try {
                onComplete.run();
            } catch (Throwable e) {
                Exceptions.throwIfFatal(e);
                RxJavaPlugins.onError(e);
            }
        }
    }
    //------------------------------FlowableSubscriber-------------------------------------

    //------------------------------Subscription-------------------------------------
    @Override
    public void dispose() {
        cancel();
    }

    @Override
    public boolean isDisposed() {
        return get() == SubscriptionHelper.CANCELLED;
    }
    //------------------------------Subscription-------------------------------------

    //------------------------------Disposable-------------------------------------
    @Override
    public void request(long n) {
        get().request(n);
    }

    @Override
    public void cancel() {
        SubscriptionHelper.cancel(this);
    }
    //------------------------------Disposable-------------------------------------

    //------------------------------LambdaConsumerIntrospection-------------------------------------
    @Override
    public boolean hasCustomOnError() {
        return onError != Functions.ON_ERROR_MISSING;
    }
    //------------------------------LambdaConsumerIntrospection-------------------------------------
}
